This pull request tries to reduce the tail latency of publishing to a stream.
Below are results of publishing to a single stream, on a 3 node cluster deployed on a single machine. The table shows throughput in msg/s and various latency measures (min, 50th percentile, 90th percentile and so on) in microseconds. In each experiment the client runs for 60 seconds.

Without the optimization in this pull request we have the following results:

Under low load, single client publishing one message at a time, latency varies between 61 microseconds, up to ~29 milliseconds for unlucky messages. Under high load, single client publishing 500 messages at a time, latency goes from ~1ms all the way to ~46ms.

One cause for the high tail latency is due to periodic snapshotting of the stream. Taking a stream snapshot involves writing a file and sync it to disk, and doing it in a safe way requires more than one call to fsync (710407c).
During this time, the monitorStream goroutine is blocked and can't process incoming messages. The above patch "moves" the writing and syncing to disk, off the stream's main thread.

With the changes in this pull request, tail latency improves significantly:

I used the "lats" benchmark client: https://github.com/synadia-labs/lats

@claude and @codex please review!

Claude finished @neilalexander's task in 4m 0s —— View job

Code Review: Asynchronous Stream Snapshots

• Read the PR diff and changed files
• Analyze the async implementation for correctness
• Check for race conditions and error handling
• Post review feedback

Summary

This PR moves the file write and Raft log compaction of stream snapshots off the monitorStream goroutine and into a separate background goroutine, using the existing RaftNodeCheckpoint interface introduced for meta-layer async snapshots. The approach is sound, the benchmarking shows meaningful tail-latency improvement, and the overall design mirrors the already-reviewed meta-layer async snapshot code.

Observations

Concurrency and mutex usage — looks correct

snapMu is used consistently to protect snapshotting, fallbackSnapshot, and failedSnapshots. The handleInstallResult closure is always called under snapMu, whether from the main goroutine (fallback path) or the async goroutine. The locally captured curState in doSnapshot is safe because only one snapshot can run at a time (guarded by snapshotting).

Behavior change: errCatchupsRunning at CreateSnapshotCheckpoint time

In the old code, errCatchupsRunning returned from n.InstallSnapshot was silently ignored — no warning, no failedSnapshots increment:

// old
} else if err != errNoSnapAvailable && err != errNodeClosed && err != errCatchupsRunning {
    s.RateLimitWarnf(...)
    failedSnapshots++
}

In the new code, errCatchupsRunning can come from CreateSnapshotCheckpoint, and the guard only excludes errNoSnapAvailable and errNodeClosed:

// new (server/jetstream_cluster.go ~L3020)
if err != errNoSnapAvailable && err != errNodeClosed {
    s.Warnf("Failed to install snapshot for '%s > %s' [%s]: %v", ...)
    failedSnapshots++
}

This means that while a catchup is in progress, each snapshot tick will log a warning and increment failedSnapshots — eventually causing forceSnapshot = failedSnapshots > 4 to kick in and override the catchup check. The meta-layer snapshot code has the same pattern (line ~1554), so this appears to be an intentional alignment rather than a bug, but it is a change in observable behaviour worth confirming. Fix this →

Shutdown with an in-flight async snapshot

On <-s.quitCh / <-mqch, the code sets fallbackSnapshot = true and then calls doSnapshot(false). If an async snapshot is already running, doSnapshot returns immediately (because snapshotting == true). The async goroutine is registered with s.grWG via s.startGoRoutine, so the server will wait for it to complete. This is correct and safe.

Redundant c.Abort() in handleInstallResult

When c.InstallSnapshot(snap) returns an error, handleInstallResult calls c.Abort() to reset n.snapshotting. However, checkpoint.InstallSnapshot already resets n.snapshotting unconditionally via defer func() { n.snapshotting = false }() (raft.go:1488). The c.Abort() call is a no-op but harmless. Worth a comment clarifying this is intentional defensive cleanup, or removing it.

Double lock/unlock in the timer case (minor)

// server/jetstream_cluster.go ~L3366
case <-t.C:
    snapMu.Lock()
    forceIfFailed := failedSnapshots > 0
    snapMu.Unlock()
    doSnapshot(forceIfFailed)  // acquires snapMu again

forceIfFailed is read under one critical section and then used in a second. The value could be stale, but doSnapshot re-checks the same condition internally under snapMu, so there's no correctness issue — just a slightly redundant lock cycle. Not worth changing.

startGoRoutine failure handling

If the server is already shutting down and startGoRoutine returns false, snapshotting is reset to false and c.Abort() is called. This correctly keeps the monitorStream and Raft states in sync. The subsequent snapshot attempt on shutdown (doSnapshot(false) from the quitCh/mqch cases) will see snapshotting == false and run the blocking fallback. Well handled.

Verdict

The implementation is correct and safe. The only item worth a second look is the errCatchupsRunning handling change — either it's intentional alignment with the meta-layer behaviour, or the condition should also exclude errCatchupsRunning to preserve the pre-existing silent-ignore semantics. Everything else looks good, and the latency numbers are compelling.

needs rebase?

needs rebase?

done

@claude review again

Claude finished @sciascid's task in 3m 53s —— View job

Code Review: Asynchronous Stream Snapshots (re-review)

• Read PR diff and changed files
• Analyze async implementation for correctness
• Check open issues from previous reviews
• Post review feedback

Summary

The implementation is sound and the design is clean. The async goroutine is correctly bounded by s.grWG, shutdown paths set fallbackSnapshot = true before calling doSnapshot, and mutex usage is consistent throughout. The latency numbers are compelling. I've re-examined all open items from prior reviews and the two remaining Codex comments.

Open Items Revisited

1. errCatchupsRunning at CreateSnapshotCheckpoint time (server/jetstream_cluster.go:3015)

This was flagged in both the previous Claude review and by Codex. The old n.InstallSnapshot path explicitly excluded errCatchupsRunning from failedSnapshots:

// old
} else if err != errNoSnapAvailable && err != errNodeClosed && err != errCatchupsRunning {
    failedSnapshots++
}

The new code does not exclude it at checkpoint creation, so each tick during a catchup will log a warning and increment failedSnapshots. This will eventually force forceSnapshot = failedSnapshots > 4, overriding the catchup guard.

sciascid's response ("This would previously happen on the n.InstallSnapshot() call") is not quite right — the old code did explicitly silence errCatchupsRunning at that path. This is a real behaviour change. However, inspecting the meta-layer counterpart at server/jetstream_cluster.go:1563 confirms it also does not exclude errCatchupsRunning from the checkpoint error path, so this is deliberate alignment with the meta layer rather than an oversight. The forced-snapshot safety valve is intentional design.

If the old silent-ignore behaviour for streams is preferred, the fix is one line: Fix this →
Otherwise, a comment clarifying the intentional change would help future readers.

2. Catchup guard only at checkpoint creation, not at install time (server/jetstream_cluster.go:3013)

Codex flagged that a catchup could start in the window between CreateSnapshotCheckpoint and c.InstallSnapshot. This is a real theoretical window. sciascid's response that the fix belongs in raft code and would affect meta snapshots too is correct — checkpoint.InstallSnapshot doesn't re-check len(n.progress). The practical impact is small (a catchup starting in that narrow window would have very little work in progress), and fixing it properly requires raft-level changes that affect both meta and stream layers. Agreed this should be tracked separately.

3. Ticker t.Reset() from async goroutine after monitorStream exits (server/jetstream_cluster.go:3052, 3069)

handleInstallResult calls t.Reset(...) from the async goroutine. If monitorStream has already exited and run defer t.Stop(), this re-arms a stopped ticker. sciascid's response is correct: t is captured by the closure, which is held by the goroutine. After the goroutine exits, there are no more references and the ticker is GC'd. The time.Ticker type is internally thread-safe, so there's no data race. The re-armed ticker may fire once with no consumer, but it will not accumulate. Not a real concern.

4. Redundant c.Abort() in handleInstallResult error path (server/jetstream_cluster.go:3057)

checkpoint.InstallSnapshot unconditionally resets n.snapshotting = false via defer (raft.go:1489–1491). The subsequent c.Abort() in handleInstallResult is therefore a no-op. Harmless. A brief comment would clarify intent for future readers.

5. Shutdown with in-flight async snapshot

When quitCh/mqch fires while an async snapshot goroutine is running, doSnapshot(false) returns immediately (because snapshotting == true) and monitorStream exits. The async goroutine continues via grWG and will complete (successfully or via errSnapAborted if the raft node closes first). If it fails, there is no second blocking-fallback attempt since the monitor has exited — but the raft log remains durable and the next startup replays from the prior snapshot + log. This is acceptable for a shutdown path.

Verdict

The implementation is correct and safe. The only item worth explicit confirmation before merge is whether the errCatchupsRunning behaviour change at CreateSnapshotCheckpoint time is intentional (aligned with meta layer) or should preserve the old silent-ignore semantics. Everything else is either intentionally aligned with the meta-layer pattern or harmless. The latency improvement justifies the change.

The only item worth explicit confirmation before merge is whether the errCatchupsRunning behaviour change at CreateSnapshotCheckpoint time is intentional

@claude We should eventually force CreateSnapshotCheckpoint if it keeps failing. So yes, this change is intentional.

Claude finished @sciascid's task in 47s —— View job

• Acknowledge sciascid's clarification on errCatchupsRunning behavior

Clarification Acknowledged

The errCatchupsRunning behavior change at CreateSnapshotCheckpoint is confirmed intentional. The code comment at server/jetstream_cluster.go:3009–3011 already documents the rationale:

// If we had a significant number of failed snapshots, start relaxing Raft-layer checks
// to force it through. We might have been catching up a peer for a long period, and this
// protects our log size from growing indefinitely.
forceSnapshot := failedSnapshots > 4

The design is: each errCatchupsRunning tick increments failedSnapshots and shortens the retry interval, so that after 5 consecutive failures the next attempt sets forceSnapshot = true and pushes the snapshot through regardless. This is the intended safety valve to bound log growth during extended catchups — a deliberate improvement over the old silent-ignore behaviour.

All open items from prior reviews are now resolved:

The implementation is correct and safe. Ready to merge.